import java.io.*;
import java.sql.*;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * MySQL到KWDB数据库数据同步工具
 * 功能：将数据从MySQL数据库同步到KWDB数据库
 * 特性：支持单表同步、通配符多表同步、智能增量同步（基于主键最大值判断）
 */
public class MySQLToKWDBDataSync {
    
    // 配置文件路径 - 改为相对路径
    private static final String CONFIG_FILE_PATH = "./config/config.txt";
    private static final String LOG_FILE_PATH = "./logs/sync_log.txt";

    // 配置项常量
    private static final String SOURCE_IP = "source.ip";
    private static final String SOURCE_PORT = "source.port";
    private static final String SOURCE_USERNAME = "source.username";
    private static final String SOURCE_PASSWORD = "source.password";
    private static final String SOURCE_DATABASE = "source.database";
    private static final String TARGET_IP = "target.ip";
    private static final String TARGET_PORT = "target.port";
    private static final String TARGET_USERNAME = "target.username";
    private static final String TARGET_PASSWORD = "target.password";
    private static final String TARGET_DATABASE = "target.database";
    private static final String SYNC_TABLES = "sync.tables";
    private static final String COLUMNS_MAPPING = "columns.mapping";
    private static final String WHERE_CONDITIONS = "where.conditions";
    private static final String BATCH_SIZE = "batch.size";
    private static final String PRIMARY_KEY = "primary.key"; // 新增：主键字段配置

    // 日志记录器
    private static PrintWriter logWriter;
    
    /**
     * 主方法 - 程序入口点
     */
    public static void main(String[] args) {
        System.out.println("=== MySQL到KWDB数据库数据同步工具开始运行 ===");
        System.out.println("配置文件路径: " + CONFIG_FILE_PATH);
        
        try {
            // 初始化日志系统
            initLogSystem();
            log("程序开始运行");
            
            // 第1步：加载MySQL和KWDB JDBC驱动程序
            Class.forName("com.mysql.cj.jdbc.Driver");
            System.out.println("✓ MySQL JDBC驱动程序加载成功");
            log("MySQL JDBC驱动程序加载成功");
            
            Class.forName("com.kaiwudb.Driver");
            System.out.println("✓ KWDB JDBC驱动程序加载成功");
            log("KWDB JDBC驱动程序加载成功");
            
            // 第2步：读取并解析配置文件
            Properties config = readConfigFile(CONFIG_FILE_PATH);
            System.out.println("✓ 配置文件解析成功");
            log("配置文件解析成功");
            
            // 第3步：建立源数据库（MySQL）和目标数据库（KWDB）的连接
            Connection sourceConn = createMySQLConnection(config);
            Connection targetConn = createKWDBConnection(config);
            System.out.println("✓ 数据库连接建立成功");
            log("数据库连接建立成功");
            
            // 第4步：解析同步配置，获取需要同步的表映射关系
            Map<String, String> tableMappings = parseTableMappings(config.getProperty(SYNC_TABLES));
            System.out.println("✓ 解析出 " + tableMappings.size() + " 个表需要同步");
            log("解析出 " + tableMappings.size() + " 个表需要同步");
            
            // 第5步：处理通配符配置（database.* 模式）
            if (tableMappings.containsKey("*")) {
                List<String> allTables;
                String specificDatabase = tableMappings.get("wildcard.db"); // 获取数据库名

                if (specificDatabase != null && !specificDatabase.isEmpty()) {
                   // 获取指定数据库下的所有表
                     allTables = getTablesFromDatabase(sourceConn, specificDatabase);
                } else {
                   // 否则获取当前连接数据库的所有表
                    allTables = getAllMySQLTables(sourceConn);
                }

                tableMappings.clear();
                for (String table : allTables) {
                    tableMappings.put(table, table); // 同名同步
                }
                System.out.println("✓ 通配符模式，发现 " + allTables.size() + " 个表需要同步");
                log("通配符模式，发现 " + allTables.size() + " 个表需要同步");
            }
            
            // 第6步：解析列映射配置
            Map<String, Map<String, String>> columnMappings = parseColumnMappings(config.getProperty(COLUMNS_MAPPING, ""));
            
            // 第7步：解析数据过滤条件
            Map<String, String> whereConditions = parseWhereConditions(config.getProperty(WHERE_CONDITIONS, ""));
            
            // 第7.5步：解析主键字段配置（新增）
            Map<String, String> primaryKeys = parsePrimaryKeys(config.getProperty(PRIMARY_KEY, ""));

            // 第8步：获取批量大小配置
            int batchSize = Integer.parseInt(config.getProperty(BATCH_SIZE, "1000"));
            
            // 第9步：执行数据同步
            executeDataSync(sourceConn, targetConn, tableMappings, columnMappings, whereConditions, batchSize, primaryKeys);
            
            // 第10步：关闭数据库连接
            sourceConn.close();
            targetConn.close();
            System.out.println("✓ 数据库连接已关闭");
            log("数据库连接已关闭");
            
            System.out.println("=== 数据同步完成 ===");
            log("数据同步完成");
            
        } catch (Exception e) {
            // 异常处理：打印错误信息
            String errorMsg = "程序执行出错: " + e.getMessage();
            System.err.println("❌ " + errorMsg);
            logError("程序执行出错", e);
            e.printStackTrace();
        } finally {
            // 关闭日志系统
            closeLogSystem();
        }
    }
    
    /**
     * 初始化日志系统
     */
    private static void initLogSystem() throws IOException {
        File logFile = new File(LOG_FILE_PATH);
        // 创建目录（如果不存在）
        logFile.getParentFile().mkdirs();
        // 使用追加模式，这样不会覆盖之前的日志
        logWriter = new PrintWriter(new FileWriter(logFile, true));
        log("=== 数据同步日志开始 ===");
    }
    
    /**
     * 记录普通日志
     */
    private static void log(String message) {
        String timestamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new java.util.Date());
        String logMessage = "[" + timestamp + "] INFO: " + message;
        
        // 输出到控制台
        System.out.println(logMessage);
        
        // 写入日志文件
        if (logWriter != null) {
            logWriter.println(logMessage);
            logWriter.flush(); // 立即刷新，确保日志被写入
        }
    }
    
    /**
     * 记录错误日志
     */
    private static void logError(String context, Exception e) {
        String timestamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new java.util.Date());
        String errorMessage = "[" + timestamp + "] ERROR: " + context + " - " + e.getMessage();
        
        // 输出到控制台
        System.err.println(errorMessage);
        
        // 写入日志文件
        if (logWriter != null) {
            logWriter.println(errorMessage);
            // 记录堆栈跟踪
            e.printStackTrace(logWriter);
            logWriter.flush();
        }
    }
    
    /**
     * 记录详细错误信息（包含表名和具体数据信息）
     */
    private static void logDataError(String tableName, String context, String dataInfo, Exception e) {
        String timestamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new java.util.Date());
        String errorMessage = "[" + timestamp + "] DATA_ERROR: 表[" + tableName + "] " + context;
        if (dataInfo != null && !dataInfo.isEmpty()) {
            errorMessage += " - 数据信息: " + dataInfo;
        }
        errorMessage += " - 错误: " + e.getMessage();
        
        // 输出到控制台
        System.err.println(errorMessage);
        
        // 写入日志文件
        if (logWriter != null) {
            logWriter.println(errorMessage);
            // 记录堆栈跟踪
            e.printStackTrace(logWriter);
            logWriter.flush();
        }
    }
    
    /**
     * 关闭日志系统
     */
    private static void closeLogSystem() {
        if (logWriter != null) {
            log("=== 数据同步日志结束 ===");
            logWriter.close();
        }
    }
    
    /**
     * 读取配置文件
     */
    private static Properties readConfigFile(String filePath) throws IOException {
        Properties props = new Properties();
        File configFile = new File(filePath);
        if (!configFile.exists()) {
            throw new FileNotFoundException("配置文件不存在: " + configFile.getAbsolutePath());
        }
        
        try (FileInputStream fis = new FileInputStream(configFile)) {
            props.load(fis);
        }
        
        validateConfig(props);
        
        return props;
    }
    
    /**
     * 验证配置完整性
     */
    private static void validateConfig(Properties props) {
        String[] requiredKeys = {SOURCE_IP, SOURCE_PORT, SOURCE_USERNAME, SOURCE_PASSWORD, SOURCE_DATABASE,
                                TARGET_IP, TARGET_PORT, TARGET_USERNAME, TARGET_PASSWORD, TARGET_DATABASE,
                                SYNC_TABLES};
        
        for (String key : requiredKeys) {
            if (!props.containsKey(key) || props.getProperty(key).trim().isEmpty()) {
                throw new IllegalArgumentException("缺少必要的配置项: " + key);
            }
        }
    }
    
    /**
     * 创建MySQL数据库连接
     */
    private static Connection createMySQLConnection(Properties config) throws SQLException {
        String url = String.format("jdbc:mysql://%s:%s/%s?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai",
                config.getProperty(SOURCE_IP),
                config.getProperty(SOURCE_PORT),
                config.getProperty(SOURCE_DATABASE));
        
        String username = config.getProperty(SOURCE_USERNAME);
        String password = config.getProperty(SOURCE_PASSWORD);
        
        log("连接MySQL数据库: " + url);
        return DriverManager.getConnection(url, username, password);
    }
    
    /**
     * 创建KWDB数据库连接
     */
    private static Connection createKWDBConnection(Properties config) throws SQLException {
        // KWDB连接URL格式，根据实际KWDB的JDBC URL格式调整
        String url = String.format("jdbc:kaiwudb://%s:%s/%s",
                config.getProperty(TARGET_IP),
                config.getProperty(TARGET_PORT),
                config.getProperty(TARGET_DATABASE));
        
        String username = config.getProperty(TARGET_USERNAME);
        String password = config.getProperty(TARGET_PASSWORD);
        
        log("连接KWDB数据库: " + url);
        return DriverManager.getConnection(url, username, password);
    }
    
    /**
     * 解析表映射配置
     */
    private static Map<String, String> parseTableMappings(String tablesConfig) {
        Map<String, String> mappings = new LinkedHashMap<>();
        
        String[] tablePairs = tablesConfig.split(",");
        for (String pair : tablePairs) {
            pair = pair.trim();
            if (pair.isEmpty()) continue;

            // 处理通配符模式：database.*
            if (pair.endsWith(".*")) {
                String databasePrefix = pair.substring(0, pair.length() - 2); // 提取数据库名
                mappings.put("*", "*"); // 标记为通配符模式
                mappings.put("wildcard.db", databasePrefix); // 特殊标记数据库
                return mappings; // 通配符优先，直接返回
            }
            // 处理显式映射：source_table->target_table
            else if (pair.contains("->")) {
                String[] parts = pair.split("->", 2); // 只分割一次
                String sourceTable = parts[0].trim();
                String targetTable = parts[1].trim();

                // 如果源表包含 .，去掉数据库前缀，只保留表名
                if (sourceTable.contains(".")) {
                    sourceTable = sourceTable.substring(sourceTable.lastIndexOf('.') + 1);
                }
                if (targetTable.contains(".")) {
                    targetTable = targetTable.substring(targetTable.lastIndexOf('.') + 1);
                }

                mappings.put(sourceTable, targetTable);
            }
            // 旧逻辑：直接用表名（同名）
            else {
                mappings.put(pair, pair);
            }
        }
        
        return mappings;
    }
    
    /**
     * 获取MySQL数据库中的所有表
     */
    private static List<String> getAllMySQLTables(Connection conn) throws SQLException {
        List<String> tables = new ArrayList<>();
        
        DatabaseMetaData metaData = conn.getMetaData();
        try (ResultSet rs = metaData.getTables(null, null, "%", new String[]{"TABLE"})) {
            while (rs.next()) {
                tables.add(rs.getString("TABLE_NAME"));
            }
        }
        
        return tables;
    }
    
    private static List<String> getTablesFromDatabase(Connection conn, String databaseName) throws SQLException {
        List<String> tables = new ArrayList<>();
        DatabaseMetaData metaData = conn.getMetaData();
        try (ResultSet rs = metaData.getTables(databaseName, null, "%", new String[]{"TABLE"})) {
            while (rs.next()) {
                tables.add(rs.getString("TABLE_NAME"));
            }
        }
        return tables;
    }

    /**
     * 解析列映射配置
     */
    private static Map<String, Map<String, String>> parseColumnMappings(String columnsConfig) {
        Map<String, Map<String, String>> mappings = new HashMap<>();
        
        if (columnsConfig == null || columnsConfig.trim().isEmpty()) {
            return mappings;
        }
        
        String[] tableMappings = columnsConfig.split(";");
        for (String tableMapping : tableMappings) {
            String[] parts = tableMapping.split(":");
            if (parts.length == 2) {
                String tableName = parts[0].trim();
                Map<String, String> columnMap = new LinkedHashMap<>();
                
                String[] columnPairs = parts[1].split(",");
                for (String columnPair : columnPairs) {
                    if (columnPair.contains("->")) {
                        String[] columnParts = columnPair.split("->");
                        if (columnParts.length == 2) {
                            columnMap.put(columnParts[0].trim(), columnParts[1].trim());
                        }
                    }
                }
                
                mappings.put(tableName, columnMap);
            }
        }
        
        return mappings;
    }
    
    /**
     * 解析WHERE条件配置
     */
    private static Map<String, String> parseWhereConditions(String whereConfig) {
        Map<String, String> conditions = new HashMap<>();
        
        if (whereConfig == null || whereConfig.trim().isEmpty()) {
            return conditions;
        }
        
        String[] tableConditions = whereConfig.split(";");
        for (String tableCondition : tableConditions) {
            String[] parts = tableCondition.split(":");
            if (parts.length == 2) {
                conditions.put(parts[0].trim(), parts[1].trim());
            }
        }
        
        return conditions;
    }

    /**
     * 解析主键字段配置（新增）
     */
    private static Map<String, String> parsePrimaryKeys(String primaryKeyConfig) {
        Map<String, String> pkMap = new HashMap<>();
        if (primaryKeyConfig == null || primaryKeyConfig.trim().isEmpty()) {
            return pkMap;
        }

        String[] pairs = primaryKeyConfig.split(",");
        for (String pair : pairs) {
            if (pair.contains("->")) {
                String[] parts = pair.split("->", 2);
                pkMap.put(parts[0].trim(), parts[1].trim());
            } else {
                // 默认主键为 id
                pkMap.put(pair.trim(), "id");
            }
        }
        return pkMap;
    }
    
    /**
     * 执行数据同步的核心方法
     */
    private static void executeDataSync(Connection sourceConn, Connection targetConn,
                                      Map<String, String> tableMappings,
                                      Map<String, Map<String, String>> columnMappings,
                                      Map<String, String> whereConditions,
                                      int batchSize,
                                      Map<String, String> primaryKeys) throws SQLException { // 新增 primaryKeys 参数
        
        System.out.println("\n=== 开始数据同步 ===");
        log("开始数据同步，共" + tableMappings.size() + "个表");
        
        long totalStartTime = System.currentTimeMillis();
        int totalTables = tableMappings.size();
        int currentTable = 0;
        long totalRecords = 0;
        int errorTables = 0; // 记录出错的表数量
        
        // 遍历所有需要同步的表
        for (Map.Entry<String, String> tableEntry : tableMappings.entrySet()) {
            currentTable++;
            String sourceTable = tableEntry.getKey();
            String targetTable = tableEntry.getValue();
            
            System.out.println("\n--- 同步表 " + currentTable + "/" + totalTables + ": " + 
                             sourceTable + " → " + targetTable + " ---");
            log("开始同步表 " + currentTable + "/" + totalTables + ": " + sourceTable + " → " + targetTable);
            
            try {
                // 同步单个表的数据
                long tableRecords = syncSingleTable(sourceConn, targetConn, sourceTable, targetTable,
                                                  columnMappings.get(sourceTable), 
                                                  whereConditions.get(sourceTable), batchSize, primaryKeys);
                totalRecords += tableRecords;
                log("表 " + sourceTable + " 同步完成，共 " + tableRecords + " 条记录");
                
            } catch (Exception e) {
                // 表级别错误：记录错误但继续同步下一个表
                errorTables++;
                String errorMsg = "同步表 " + sourceTable + " 时出错，跳过此表继续执行";
                System.err.println("❌ " + errorMsg);
                logError(errorMsg, e);
            }
        }
        
        long totalEndTime = System.currentTimeMillis();
        long totalDuration = totalEndTime - totalStartTime;
        
        System.out.println("\n=== 同步完成汇总 ===");
        String summary = String.format("同步表数量: %d, 成功: %d, 失败: %d, 总记录数: %d, 总耗时: %d 毫秒", 
                totalTables, totalTables - errorTables, errorTables, totalRecords, totalDuration);
        System.out.println(summary);
        log("同步完成汇总: " + summary);
        
        // 添加性能统计
        if (totalDuration > 0) {
            double recordsPerSecond = (double) totalRecords / totalDuration * 1000;
            String performanceMsg = String.format("同步性能: %.2f 条/秒", recordsPerSecond);
            System.out.println(performanceMsg);
            log(performanceMsg);
        }
    }
    
    /**
     * 同步单个表的数据（支持智能起始点：全量/增量）
     */
    private static long syncSingleTable(Connection sourceConn, Connection targetConn,
                                      String sourceTable, String targetTable,
                                      Map<String, String> columnMapping,
                                      String whereCondition, int batchSize,
                                      Map<String, String> primaryKeys) throws SQLException { // 新增参数

        long startTime = System.currentTimeMillis();
        long recordCount = 0;

        try {
            // 获取主键字段名（默认为 id）
            String primaryKey = primaryKeys.getOrDefault(sourceTable, "id");
            log("使用主键字段: " + primaryKey + " 进行同步起点判断");

            // 第一步：查询目标表的最大主键值
            Long maxId = getMaxPrimaryKeyValue(targetConn, targetTable, primaryKey);
            boolean isIncremental = maxId != null && maxId > 0;

            log("目标表 [" + targetTable + "] 当前最大主键值: " + (maxId == null ? "null (全量同步)" : maxId + " (增量同步)"));

            // 第二步：获取源表（MySQL）的列信息
            Map<String, String> sourceColumns = getMySQLTableColumns(sourceConn, sourceTable);
            if (sourceColumns.isEmpty()) {
                String errorMsg = "源表不存在或没有列: " + sourceTable;
                System.out.println("❌ " + errorMsg);
                log(errorMsg);
                return 0;
            }
            
            // 第三步：获取目标表（KWDB）的列信息
            Map<String, String> targetColumns = getKWDBTableColumns(targetConn, targetTable);
            if (targetColumns.isEmpty()) {
                String errorMsg = "目标表不存在或没有列: " + targetTable;
                System.out.println("❌ " + errorMsg);
                log(errorMsg);
                return 0;
            }
            
            // 第四步：确定实际要同步的列
            List<String> finalSourceColumns = new ArrayList<>();
            List<String> finalTargetColumns = new ArrayList<>();
            
            if (columnMapping != null && !columnMapping.isEmpty()) {
                for (Map.Entry<String, String> entry : columnMapping.entrySet()) {
                    if (sourceColumns.containsKey(entry.getKey()) && 
                        targetColumns.containsKey(entry.getValue())) {
                        finalSourceColumns.add(entry.getKey());
                        finalTargetColumns.add(entry.getValue());
                    }
                }
            } else {
                for (String sourceCol : sourceColumns.keySet()) {
                    if (targetColumns.containsKey(sourceCol)) {
                        finalSourceColumns.add(sourceCol);
                        finalTargetColumns.add(sourceCol);
                    }
                }
            }
            
            if (finalSourceColumns.isEmpty()) {
                String errorMsg = "没有找到可同步的列";
                System.out.println("❌ " + errorMsg);
                log(errorMsg);
                return 0;
            }
            
            System.out.println("同步列: " + finalSourceColumns + " → " + finalTargetColumns);
            System.out.println("源表列类型: " + sourceColumns);
            System.out.println("目标表列类型: " + targetColumns);
            log("同步列: " + finalSourceColumns + " → " + finalTargetColumns);

            // 第五步：构建查询SQL和插入SQL
            String selectSql = buildSelectSql(sourceTable, finalSourceColumns, whereCondition, isIncremental ? primaryKey : null, maxId);
            String insertSql = buildInsertSql(targetTable, finalTargetColumns);
            
            System.out.println("查询SQL: " + selectSql);
            System.out.println("插入SQL: " + insertSql);

            // 第六步：执行数据同步
            recordCount = executeTableSync(sourceConn, targetConn, selectSql, insertSql, 
                                         finalSourceColumns.size(), batchSize, sourceTable);
            
            long endTime = System.currentTimeMillis();
            long duration = endTime - startTime;
            
            String successMsg = String.format("表同步完成: %d 条记录, 耗时: %d 毫秒, 速度: %.2f 条/秒",
                    recordCount, duration, 
                    recordCount > 0 ? (double)recordCount / duration * 1000 : 0);
            System.out.println("✓ " + successMsg);
            log(successMsg);
            
        } catch (SQLException e) {
            String errorMsg = "同步表 " + sourceTable + " 时出错";
            System.err.println("❌ " + errorMsg + ": " + e.getMessage());
            logError(errorMsg, e);
            throw e;
        }
        
        return recordCount;
    }

    /**
     * 获取目标表中某主键字段的最大值（用于判断是否需要增量同步）
     */
    private static Long getMaxPrimaryKeyValue(Connection conn, String tableName, String primaryKey) {
        String sql = "SELECT MAX(" + primaryKey + ") FROM " + tableName;
        try (PreparedStatement stmt = conn.prepareStatement(sql);
             ResultSet rs = stmt.executeQuery()) {
            if (rs.next()) {
                Object value = rs.getObject(1);
                if (value == null) {
                    log("目标表 [" + tableName + "] 为空或无数据 (MAX(" + primaryKey + ") IS NULL)");
                    return null;
                } else {
                    Number num = (Number) value;
                    Long maxVal = num.longValue();
                    log("目标表 [" + tableName + "] 最大主键值为: " + maxVal);
                    return maxVal;
                }
            }
        } catch (SQLException e) {
            log("查询目标表最大主键失败: " + sql + ", 错误: " + e.getMessage());
            // 如果查询失败，为了安全起见，可以当作全量同步处理
            log("查询失败，将按全量同步处理");
            return null;
        }
        return null; // 理论上不应该到达这里
    }
    
    /**
     * 获取MySQL表的列信息
     */
    private static Map<String, String> getMySQLTableColumns(Connection conn, String tableName) throws SQLException {
        Map<String, String> columns = new LinkedHashMap<>();
        
        DatabaseMetaData metaData = conn.getMetaData();
        try (ResultSet rs = metaData.getColumns(null, null, tableName, "%")) {
            while (rs.next()) {
                String columnName = rs.getString("COLUMN_NAME");
                String columnType = rs.getString("TYPE_NAME");
                columns.put(columnName, columnType);
            }
        }
        
        return columns;
    }
    
    /**
     * 获取KWDB表的列信息
     */
    private static Map<String, String> getKWDBTableColumns(Connection conn, String tableName) throws SQLException {
        Map<String, String> columns = new LinkedHashMap<>();
        
        try {
            DatabaseMetaData metaData = conn.getMetaData();
            try (ResultSet rs = metaData.getColumns(null, null, tableName, "%")) {
                while (rs.next()) {
                    String columnName = rs.getString("COLUMN_NAME");
                    String columnType = rs.getString("TYPE_NAME");
                    columns.put(columnName, columnType);
                }
            }
        } catch (SQLException e) {
            // 如果KWDB的元数据获取方式不同，尝试使用SHOW COLUMNS语句
            log("使用元数据获取KWDB列信息失败，尝试使用SHOW COLUMNS: " + e.getMessage());
            try (Statement stmt = conn.createStatement();
                 ResultSet rs = stmt.executeQuery("SHOW COLUMNS FROM " + tableName)) {
                while (rs.next()) {
                    String columnName = rs.getString(1); // 第一列是列名
                    String columnType = rs.getString(2); // 第二列是数据类型
                    columns.put(columnName, columnType);
                }
            }
        }
        
        return columns;
    }

    /**
     * 构建查询SQL（支持增量同步条件）
     */
    private static String buildSelectSql(String tableName,
                                       List<String> columns,
                                       String whereCondition,
                                       String primaryKey, // 可为 null
                                       Long lastMaxId) { // 可为 null

        StringBuilder sql = new StringBuilder();
        sql.append("SELECT ").append(String.join(", ", columns))
           .append(" FROM ").append(tableName);

        List<String> conditions = new ArrayList<>();
        if (whereCondition != null && !whereCondition.trim().isEmpty()) {
            conditions.add("(" + whereCondition + ")");
        }

        // 如果是增量模式，加上主键 > maxId 的条件
        if (primaryKey != null && lastMaxId != null) {
            conditions.add(primaryKey + " > " + lastMaxId);
        }

        if (!conditions.isEmpty()) {
            sql.append(" WHERE ").append(String.join(" AND ", conditions));
        }

        return sql.toString();
    }
    
    /**
     * 构建插入SQL
     */
    private static String buildInsertSql(String tableName, List<String> columns) {
        String columnsStr = String.join(", ", columns);
        String placeholders = String.join(", ", Collections.nCopies(columns.size(), "?"));
        
        return "INSERT INTO " + tableName + " (" + columnsStr + ") VALUES (" + placeholders + ")";
    }
    
    /**
     * 执行表数据同步的核心方法（针对KWDB优化的高性能写入）
     */
    private static long executeTableSync(Connection sourceConn, Connection targetConn,
                                       String selectSql, String insertSql, 
                                       int columnCount, int batchSize, String tableName) throws SQLException {
        
        long totalRecords = 0;
        int errorRecords = 0; // 记录出错的记录数
        
        try (PreparedStatement selectStmt = sourceConn.prepareStatement(selectSql);
             ResultSet rs = selectStmt.executeQuery()) {
            
            boolean originalAutoCommit = targetConn.getAutoCommit();
            targetConn.setAutoCommit(false);
            
            try (PreparedStatement insertStmt = targetConn.prepareStatement(insertSql)) {
                System.out.println("开始高性能批量同步到KWDB...");
                log("开始高性能批量同步表 " + tableName + " 到KWDB");
                
                int batchCount = 0;
                
                while (rs.next()) {
                    totalRecords++;
                    batchCount++;
                    
                    try {
                        // 设置插入参数
                        for (int i = 1; i <= columnCount; i++) {
                            Object value = rs.getObject(i);
                            insertStmt.setObject(i, value);
                        }
                        
                        insertStmt.addBatch();
                        
                        // 达到批处理大小时执行批处理
                        if (batchCount % batchSize == 0) {
                            int[] results = insertStmt.executeBatch();
                            targetConn.commit();
                            insertStmt.clearBatch();
                            batchCount = 0;
                            System.out.println("  已同步 " + totalRecords + " 条记录...");
                        }
                        
                    } catch (Exception e) {
                        // 单条记录错误：记录错误但继续处理下一条记录
                        errorRecords++;
                        String dataInfo = "记录号:" + totalRecords;
                        logDataError(tableName, "单条记录处理出错", dataInfo, e);
                        
                        // 继续处理下一条记录
                        continue;
                    }
                }
                
                // 执行剩余的批处理
                if (batchCount > 0) {
                    try {
                        int[] results = insertStmt.executeBatch();
                        targetConn.commit();
                    } catch (Exception e) {
                        errorRecords += batchCount;
                        logDataError(tableName, "最后批次执行出错", "批次大小:" + batchCount, e);
                        targetConn.rollback();
                    }
                }
                
                targetConn.setAutoCommit(originalAutoCommit);
                
                String completionMsg = String.format("表同步完成，共处理 %d 条记录，其中 %d 条失败", 
                        totalRecords, errorRecords);
                System.out.println("✓ " + completionMsg);
                log(completionMsg);
                
            } catch (SQLException e) {
                targetConn.rollback();
                targetConn.setAutoCommit(originalAutoCommit);
                throw e;
            }
        }
        
        return totalRecords - errorRecords; // 返回成功处理的记录数
    }
}



